package Sink

import Source.SensorReading
import org.apache.flink.api.common.serialization.SimpleStringEncoder
import org.apache.flink.core.fs.Path
import org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink
import org.apache.flink.streaming.api.scala._

/**
 * 使用Kafka作为Sink
 */
object SinkToLocalFile {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val inputPath = "src/main/resources/SensorReading"
    val inputStream = env.readTextFile(inputPath)

    //转换成样例类类型
    val dataStream = inputStream.map(
      data => {
        val arr = data.split(",")
        SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
      }
    )
    //    dataStream.writeAsCsv("src/main/resources/out.csv")
    dataStream.addSink(StreamingFileSink.forRowFormat(
      new Path("src/main/resources/out.csv"),
      new SimpleStringEncoder[SensorReading]()
    ).build())
    env.execute()
  }
}
